#include <stddef.h>
#include <stdio.h>
#include <stdint.h>
#include <stdlib.h>
#include <unistd.h>
#include <sys/ipc.h>
#include <sys/shm.h>
#include <unistd.h>
#include "golfdds.h"
#include "ring_queue.h"
#include "golfdds_shm.h"
#include <pthread.h>

#include <string.h>

#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <netinet/ip.h>
#include <arpa/inet.h>
#include <sys/select.h>
#define SHMINDEX 194
/*
中心化，需要节点各自探测需求：
订阅者需要探测发布的主题是否存在，如果存在，需要及时产生发布-订阅关系
发布者需要管理订阅者，接受并回复订阅者
基本接口就是：
1.创建节点，节点可以发布，也可以订阅主题
2.节点可以创建一个发布，发布有一个handle，可以通过此handle发布主题数据
3.节点可以创建一个订阅，可以订阅数据
所以，有以下实体需要定义：
node,topic,subscriber,publisher
node:需要有一个node才能创建subcriber/publisher
topic:subscriber/publisher处理的对象就是topic,topic对外用一个可读的字符串表达，
在node内部，如果是publisher，需要维护topic的完整定义：
1.是哪个node发布的，
2.有哪些node来订阅，
3.消息大小，
4.队列长度
为了明确他们的关系，我们以测试代码驱动开发
以GOLFDDS_为前缀：
GOLFDDS_NODE
GOLFDDS_TOPIC
GOLFDDS_SUBSCRIBER
GOLFDDS_PUBLISHER
它们对外都是一个void*,对内都是一个结构体

*/
typedef enum {
  DETACHED,
  ATTACHED,
} SUBSCRIBERSTATUS;
typedef struct _golfdds_subscriber_instance {
  double lastattachtime;
  int sockfd;
  SUBSCRIBERSTATUS status;
  char topicname[64];
  char shmpath[256];
  int shmid;
  char nodename[64];
  int msgsize;
  int queuelen;
  Ring_Queue *msgqueue;
  GOLFDDS_MSGCALLBACK callback;
  struct _golfdds_subscriber_instance *next;
} GOLFDDS_SUBSCRIBER_INSTANCE;

typedef struct
{
  char shmpath[256];
  Ring_Queue *msgqueue;
} SHMINFO;
typedef struct
{
  int count;
  SHMINFO info[32];
} SHMINFOS;
typedef struct _golfdds_publisher_instance {
  int sockfd;
  char topicname[64];
  SHMINFOS shminfos;

  int shmid;
  int msgsize;
  int queuelen;

  int subscribernum;
  GOLFDDS_MSGCALLBACK callback;
  struct _golfdds_publisher_instance *next;
} GOLFDDS_PUBLISHER_INSTANCE;
typedef struct
{
  char nodename[64];
  pthread_t nodeserverthreadid;
  int willrun;
  int sockfd;
  int serverport;
  time_t lastquerytm;

  GOLFDDS_SUBSCRIBER_INSTANCE *subscriber;
  GOLFDDS_PUBLISHER_INSTANCE *publisher;
  int willexit;
} GOLFDDS_NODE_INSTANCE;

void udpsend(int sockfd, void *msg, int len, char *ip, int port) {
  // 4、发送数据
  struct sockaddr_in recv_addr = {0};
  socklen_t addr_len = sizeof(struct sockaddr);

  // 注意输入广播地址，格式为*.*.*.255
  recv_addr.sin_family = AF_INET;            // 指定用IPV4地址
  recv_addr.sin_port = htons(port);          // 接收者的端口号
  recv_addr.sin_addr.s_addr = inet_addr(ip); // 接收者的IP地址
  // 发送数据，注意要填写接收者的地址
  int ret = sendto(sockfd, msg, len, 0,
                   (struct sockaddr *)&recv_addr, addr_len);
  if (ret > 0) {
    // printf("attach_subscriber-success: send %d bytes,%d\n", ret,sub->sockfd);
  }
}
void sendpublisherinfo(GOLFDDS_NODE_INSTANCE *inst, GOLFDDS_PUBLISHER_INSTANCE *tmppub, int port, int index) {
  PUBSUBINFO info;
  info.type = PUBINFO;
  PUBLISHERINFO *pubinfo = &info.pubinfo;
  strcpy(pubinfo->nodename, inst->nodename);
  strcpy(pubinfo->topicname, tmppub->topicname);
  strcpy(pubinfo->shmpath, tmppub->shminfos.info[index].shmpath);
  pubinfo->shmid = tmppub->shmid;
  pubinfo->msgsize = tmppub->msgsize;
  pubinfo->queuelen = tmppub->queuelen;

  udpsend(inst->sockfd, &info, sizeof(PUBSUBINFO), "127.0.0.1", port);
}
/*
void sendsubscriberinfo(GOLFDDS_NODE_INSTANCE *inst, GOLFDDS_SUBSCRIBER_INSTANCE *tmpsub, int port)
{
  PUBSUBINFO info;
  info.type = SUBSCRIBER;
  PUBLISHERINFO *pubinfo = &info.pubinfo;
  strcpy(pubinfo->nodename, inst->nodename);
  strcpy(pubinfo->topicname, tmpsub->topicname);
  info.nodeport = port;
  // strcpy(pubinfo->shmpath, tmpsub->shmpath);
  // pubinfo->shmid = tmpsub->shmid;
  // pubinfo->msgsize = tmpsub->msgsize;
  // pubinfo->queuelen = tmpsub->queuelen;

  udpsend(inst->sockfd, &info, sizeof(PUBSUBINFO), "127.0.0.1", 9527);
}
*/
static void *nodeserverthread(void *param) { // 每个节点有一个服务端udp口，可以用来接受来自其他节点的信息

  fd_set rfds;
  struct timeval tv;
  int retval;
  GOLFDDS_NODE_INSTANCE *inst = (GOLFDDS_NODE_INSTANCE *)param;
  inst->sockfd = socket(AF_INET, SOCK_DGRAM, 0);
  if (inst->sockfd == -1) {
    perror("node sockfd create fail\n");
    return 0;
  }
  socklen_t addr_len = sizeof(struct sockaddr);

  struct sockaddr_in send_addr;
  send_addr.sin_addr.s_addr = inet_addr("127.0.0.1");
  send_addr.sin_family = AF_INET; // 指定协议为IPV4地址协议
  send_addr.sin_port = htons(0);  // 端口号
  int ret = bind(inst->sockfd, (struct sockaddr *)&send_addr, addr_len);
  if (ret == -1) {
    printf("bind node sockfd fail:%d\n", inst->sockfd);
    close(inst->sockfd);
    return;
  }

  struct sockaddr_in node_addr;
  // 获取并打印本地端口
  if (getsockname(inst->sockfd, (struct sockaddr *)&node_addr, &addr_len) < 0) {
    perror("get node sockfd's port failed");
    return;
  }

  inst->serverport = ntohs(node_addr.sin_port);
  printf("node sockfd's port: %d\n", ntohs(node_addr.sin_port));

  while (inst->willrun) {
    FD_ZERO(&rfds);
    FD_SET(inst->sockfd, &rfds);
    tv.tv_sec = 1;
    tv.tv_usec = 0;

    retval = select(inst->sockfd + 1, &rfds, NULL, NULL, &tv);
    if (retval == -1) {
      perror("select");
    } else if (retval) { // 有来自其他node的信息
      // printf("some info from othernode!\n");
      char buffer[1024];
      struct sockaddr_in cliaddr;
      socklen_t addrlen = sizeof(cliaddr);

      if (FD_ISSET(inst->sockfd, &rfds)) {
        ssize_t len = recvfrom(inst->sockfd, buffer, sizeof(buffer), 0, (struct sockaddr *)&cliaddr, &addrlen);

        PUBSUBINFO *pubsubinfo = (PUBSUBINFO *)buffer;
        if (pubsubinfo->type == SUBLIST) {
          SUBSCRIBERLIST *list = &pubsubinfo->list;
          /*printf("receive subscriber list from some other node:\n");
          for (int i = 0; i < list->count; i++)
          {
            printf("%s ", &list->topics[i][0]);
          }
          printf("\n");
*/
          // 遍历本节点所有发布者
          GOLFDDS_PUBLISHER_INSTANCE *tmppub = inst->publisher;
          while (tmppub != NULL) {
            for (int i = 0; i < list->count; i++) {                      // 有其他节点查询本节点是否有想要的发布主题
              if (0 == strcmp(tmppub->topicname, &list->topics[i][0])) { // 如果有
                if (tmppub->shminfos.count >= 32) {
                  printf("topic:%s's shmcount has reach 32,just ignore publisher query!\n", tmppub->topicname);
                  break;
                }
                printf("will test to create shm,index:%d!\n", tmppub->shminfos.count);
                char tmppath[64];
                sprintf(tmppath, "/tmp/%s_%d_%s", inst->nodename, ntohs(cliaddr.sin_port), tmppub->topicname);
                int hascreated = 0;
                for (int j = 0; j < tmppub->shminfos.count; j++) {
                  if (0 == strcmp(tmppub->shminfos.info[j].shmpath, tmppath)) {
                    sendpublisherinfo(inst, tmppub, ntohs(cliaddr.sin_port), j);
                    hascreated = 1;
                  }
                }
                if (hascreated == 0) {
                  strcpy(tmppub->shminfos.info[tmppub->shminfos.count].shmpath, tmppath);
                  FILE *fp = fopen(&tmppub->shminfos.info[tmppub->shminfos.count].shmpath, "w");
                  fclose(fp);
                  printf("will create shm of topic:%s,path:%s\n", tmppub->topicname, tmppath);
                  tmppub->shminfos.info[tmppub->shminfos.count]
                      .msgqueue = create_or_attach_shm(&tmppub->shminfos.info[tmppub->shminfos.count].shmpath,
                                                       SHMINDEX, tmppub->msgsize, tmppub->queuelen, &tmppub->shmid);
                  sendpublisherinfo(inst, tmppub, ntohs(cliaddr.sin_port), tmppub->shminfos.count);
                  tmppub->shminfos.count++;
                }
                break;
              }
            }

            tmppub = tmppub->next;
          }
        } else if (pubsubinfo->type == PUBINFO) {
          // printf("recv pub info!\n");
          PUBLISHERINFO *ppubinfo = &pubsubinfo->pubinfo;

          GOLFDDS_SUBSCRIBER_INSTANCE *tmpsub = inst->subscriber;
          while (tmpsub != NULL) {
            if (0 == strcmp(tmpsub->topicname, ppubinfo->topicname)) {
              if (tmpsub->status == DETACHED) {
                strcpy(tmpsub->shmpath, ppubinfo->shmpath);
                tmpsub->shmid = ppubinfo->shmid;
                tmpsub->msgsize = ppubinfo->msgsize;
                tmpsub->queuelen = ppubinfo->queuelen;

                tmpsub->msgqueue = attach_shm(ppubinfo->shmpath, SHMINDEX, &tmpsub->shmid);
                printf("has attach topic:%s,msgqueue:0x%X\n", tmpsub->topicname, tmpsub->msgqueue);
                if (tmpsub->msgqueue != NULL)
                  tmpsub->status = ATTACHED;
              }
              break;
            }
            tmpsub = tmpsub->next;
          }
        }
        // 这里，查询本节点是否有该名称节点发布，如果有，则新建共享内存并发回相关信息

        /*     struct timeval tv;
             gettimeofday(&tv, NULL);
             double now = (double)tv.tv_sec + (double)tv.tv_usec / 1000000.0;

             PUBSUBINFO *ppubinfo = (PUBSUBINFO *)buffer;

             // 这里可以attach上去了
             if (ppubinfo->type == PUBLISHER)
             {
               printf("%f,node sockfd Received: type:%d,%s,%s,%s,%d,%d,%d\n", now, ppubinfo->type,
                      ppubinfo->pubinfo.nodename, ppubinfo->pubinfo.topicname,
                      ppubinfo->pubinfo.shmpath, ppubinfo->pubinfo.shmid, ppubinfo->pubinfo.msgsize, ppubinfo->pubinfo.queuelen);
               GOLFDDS_SUBSCRIBER_INSTANCE *tmpsub = inst->subscriber;
               while (tmpsub != NULL)
               {
                 if (0 == strcmp(tmpsub->topicname, ppubinfo->pubinfo.topicname))
                 {
                   if (tmpsub->status == DETACHED)
                   {
                     strcpy(tmpsub->shmpath, ppubinfo->pubinfo.shmpath);
                     tmpsub->shmid = ppubinfo->pubinfo.shmid;
                     tmpsub->msgsize = ppubinfo->pubinfo.msgsize;
                     tmpsub->queuelen = ppubinfo->pubinfo.queuelen;

                     tmpsub->msgqueue = attach_shm(ppubinfo->pubinfo.shmpath, ppubinfo->pubinfo.shmid);
                     printf("has attach topic:%s\n", tmpsub->topicname);
                     tmpsub->status = ATTACHED;
                   }
                   break;
                 }
                 tmpsub = tmpsub->next;
               }
             }
             else
             {
               printf("%f,node sockfd Received: type:%d,%s,%s,%s,%d,%d,%d\n", now, ppubinfo->type,
                      ppubinfo->pubinfo.nodename, ppubinfo->pubinfo.topicname,
                      ppubinfo->pubinfo.shmpath, ppubinfo->pubinfo.shmid, ppubinfo->pubinfo.msgsize, ppubinfo->pubinfo.queuelen);
               GOLFDDS_PUBLISHER_INSTANCE *tmppub = inst->publisher;
               while (tmppub != NULL)
               {
                 if (0 == strcmp(tmppub->topicname, ppubinfo->pubinfo.topicname))
                 {
                   if (tmppub->subscribernum == 0)
                     tmppub->subscribernum = 1;
                   break;
                 }
                 tmppub = tmppub->next;
               }
             }*/
      }
    }
    /* else
     {
       // printf("Timeout,%d\n", retval);
       GOLFDDS_PUBLISHER_INSTANCE *tmppub = inst->publisher;
       while (tmppub != NULL)
       {
         sendpublisherinfo(inst, tmppub, ntohs(node_addr.sin_port));
         tmppub = tmppub->next;
       }
       GOLFDDS_SUBSCRIBER_INSTANCE *tmpsub = inst->subscriber;
       while (tmpsub != NULL)
       {
         sendsubscriberinfo(inst, tmpsub, ntohs(node_addr.sin_port));
         tmpsub = tmpsub->next;
       }
     }*/
  }
}
GOLFDDS_NODE_HANDLE golfdds_create_node(const char *nodename) {
  GOLFDDS_NODE_INSTANCE *inst = (GOLFDDS_NODE_INSTANCE *)calloc(1, sizeof(GOLFDDS_NODE_INSTANCE));
  strcpy(inst->nodename, nodename);

  inst->willrun = 1;
  pthread_create(&inst->nodeserverthreadid, NULL, nodeserverthread, inst); // detect publisher
  return (GOLFDDS_NODE_HANDLE)inst;
}
GOLFDDS_SUBSCRIBER_HANDLE golfdds_create_subscriber(GOLFDDS_NODE_HANDLE node, const char *topicname, GOLFDDS_MSGCALLBACK msgcallback) { // 这里需要探测发布者，探测到就创建连接到发布者，否则就返回失败
  GOLFDDS_NODE_INSTANCE *nodeinst = (GOLFDDS_NODE_INSTANCE *)node;
  GOLFDDS_SUBSCRIBER_INSTANCE *inst = (GOLFDDS_SUBSCRIBER_INSTANCE *)calloc(1, sizeof(GOLFDDS_SUBSCRIBER_INSTANCE));
  strcpy(inst->topicname, topicname);
  strcpy(inst->nodename, nodeinst->nodename);
  inst->callback = msgcallback;
  inst->next = nodeinst->subscriber;
  nodeinst->subscriber = inst;

  return (GOLFDDS_SUBSCRIBER_HANDLE)inst;
}
GOLFDDS_PUBLISHER_HANDLE golfdds_create_publisher(GOLFDDS_NODE_HANDLE node, const char *topicname, int msgsize, int queuelen) {
  GOLFDDS_NODE_INSTANCE *nodeinst = (GOLFDDS_NODE_INSTANCE *)node;
  GOLFDDS_PUBLISHER_INSTANCE* tmppub=nodeinst->publisher;
  while(tmppub!=NULL){
	if(0==strcmp(tmppub->topicname,topicname)){
		return (GOLFDDS_PUBLISHER_HANDLE)tmppub;
	}
	tmppub=tmppub->next;
  } 
  GOLFDDS_PUBLISHER_INSTANCE *inst = (GOLFDDS_PUBLISHER_INSTANCE *)calloc(1, sizeof(GOLFDDS_PUBLISHER_INSTANCE));
  strncpy(inst->topicname, topicname, 63);
  inst->msgsize = msgsize;
  inst->queuelen = queuelen;

  inst->next = nodeinst->publisher;
  nodeinst->publisher = inst;
  return (GOLFDDS_PUBLISHER_HANDLE)inst;
}

void golfdds_publish(GOLFDDS_PUBLISHER_HANDLE publisher, void *message) {
  GOLFDDS_PUBLISHER_INSTANCE *inst = (GOLFDDS_PUBLISHER_INSTANCE *)publisher;
  for (int i = 0; i < inst->shminfos.count; i++) {
    if (inst->shminfos.info[i].msgqueue != NULL) {
      uint8_t *p = (uint8_t *)SOLO_Write_SHM(inst->shminfos.info[i].msgqueue);
      if (p) {
        // printf("has send msg of topic:%s\n", inst->topicname);
        memcpy(p, message, inst->msgsize);
        SOLO_Write_Over_SHM(inst->shminfos.info[i].msgqueue);
      } else {
        // printf("miss write queue!\n");
      }
    }
  }
}
int golfdds_subscriber_count(GOLFDDS_PUBLISHER_HANDLE publisher) {
  GOLFDDS_PUBLISHER_INSTANCE *inst = (GOLFDDS_PUBLISHER_INSTANCE *)publisher;

  return inst->shminfos.count;
}
/*
static void
attach_subscriber(GOLFDDS_SUBSCRIBER_INSTANCE *sub)
{ // 通过本地广播获取连接
  struct timeval tv;
  gettimeofday(&tv, NULL);
  double now = (double)tv.tv_sec + (double)tv.tv_usec / 1000000.0;
  if ((now - sub->lastattachtime) < 1.0)
  {
    return;
  }
  sub->lastattachtime = now;
  socklen_t addr_len = sizeof(struct sockaddr);
  struct sockaddr_in send_addr;
  send_addr.sin_addr.s_addr = inet_addr("127.0.0.1");
  send_addr.sin_family = AF_INET; // 指定协议为IPV4地址协议
  send_addr.sin_port = htons(0);  // 端口号
  // 1、建立套接字，使用IPV4网络地址，UDP协议
  if (sub->sockfd == 0 || sub->sockfd == -1)
  {
    sub->sockfd = socket(AF_INET, SOCK_DGRAM, 0);
    if (sub->sockfd == -1)
    {
      perror("attach_subscriber-socket fail");
      return;
    }

    // 3、绑定自己的IP地址和端口号（可以省略）

    int ret = bind(sub->sockfd, (struct sockaddr *)&send_addr, addr_len);
    if (ret == -1)
    {
      printf("attach_subscriber-bind fail:%d\n", sub->sockfd);
      close(sub->sockfd);
      return;
    }
  }

  // 获取并打印本地端口
  if (getsockname(sub->sockfd, (struct sockaddr *)&send_addr, &addr_len) < 0)
  {
    perror("attach_subscriber-getsockname failed");
    return;
  }

  //  printf("attach_subscriber-Local port: %d\n", ntohs(send_addr.sin_port));

  // 4、发送数据
  uint16_t port = 0; // 端口号
  char ip[20] = {0}; // IP地址
  struct sockaddr_in recv_addr = {0};

  char msg[1024] = {0}; // 数据缓冲区

  // 注意输入广播地址，格式为*.*.*.255
  recv_addr.sin_family = AF_INET;                     // 指定用IPV4地址
  recv_addr.sin_port = htons(9527);                   // 接收者的端口号
  recv_addr.sin_addr.s_addr = inet_addr("127.0.0.1"); // 接收者的IP地址

  SUBSCRIBERINFO info;
  strcpy(info.nodename, sub->nodename);
  strcpy(info.topicname, sub->topicname);
  info.subscriberport = ntohs(send_addr.sin_port);
  {
    // 发送数据，注意要填写接收者的地址
    int ret = sendto(sub->sockfd, &info, sizeof(SUBSCRIBERINFO), 0,
                     (struct sockaddr *)&recv_addr, addr_len);
    if (ret > 0)
    {
      printf("send to core ok!\n");
      // printf("attach_subscriber-success: send %d bytes,%d\n", ret,sub->sockfd);
    }
    else
    {
      printf("send to core failed!\n");
    }
  }
}
*/

void queryudpport(GOLFDDS_NODE_INSTANCE *inst, SUBSCRIBERLIST *plist) {
  // printf("############will query udp port!\n");
  FILE *fp = fopen("/proc/net/udp", "r");
  if (fp == NULL) {
    perror("fopen");
    return 1;
  }

  char line[256];
  while (fgets(line, sizeof(line), fp)) {
    unsigned int local_addr, local_port;
    uint32_t iphex;
    // 解析每行，寻找127.0.0.1的绑定情况
    if (sscanf(line, " %*s %x:%x %*s", &iphex, &local_port) == 2 &&
        iphex == 0x0100007F) {
      // printf("Local UDP port: %u\n", local_port);
      if (local_port != inst->serverport) {
        // printf("%d,will send query subscriberlist to port:%d\n", time(NULL), local_port);
        PUBSUBINFO info;
        info.list = *plist;
        info.type = SUBLIST;
        udpsend(inst->sockfd, &info, sizeof(PUBSUBINFO), "127.0.0.1", local_port);
      }
    }
  }

  fclose(fp);
  // printf("############query udp port end!\n");
}

void detectpublisher(GOLFDDS_NODE_INSTANCE *inst) { // 这里，向绑定127.0.0.1的端口们发送查询消息
  // 生成inst节点的所有订阅者信息
  // 遍历本节点的订阅者
  int index = 0;
  SUBSCRIBERLIST info;
  GOLFDDS_SUBSCRIBER_INSTANCE *tmpsub = inst->subscriber;
  while (tmpsub != NULL) {
    if (tmpsub->status != ATTACHED) {
      // 如果当前需要订阅的主题没有attach到，那么就需要探测
      if (index > 32) {
        printf("warning,only 32 topics can be contained in one udp packet!\n");
        break;
      }
      strcpy(&info.topics[index][0], tmpsub->topicname);
      index++;
    } else {
      // 但是如果已经attach，后来断开了，那怎么办？
      // 通过shmctl获取对端是否已经因为推出，导致离开了
      struct shmid_ds buf;
      int ret = shmctl(tmpsub->shmid, IPC_STAT, &buf);
      if (ret != -1) {
        // printf("got nattach:%lu of shmid:%d\n", buf.shm_nattch, tmpsub->shmid);
        if (buf.shm_nattch < 2) {
          printf("add attached.....\n");
          shmdt(tmpsub->msgqueue);
          printf("will rmid of shmid:%d\n", tmpsub->shmid);
          shmctl(tmpsub->shmid, IPC_RMID, NULL);
          tmpsub->status = DETACHED;
          if (index > 32) {
            printf("warning,only 32 topics can be contained in one udp packet!\n");
            break;
          }
          strcpy(&info.topics[index][0], tmpsub->topicname);
          index++;
        }
      }
    }
    tmpsub = tmpsub->next;
  }
  if (index == 0) {
    // printf("node has no subscriber not attached!\n");
    return;
  }
  info.count = index;
  ////////////////////////////有未attach的topic需要发送查询
  // 找到所有本地绑定到127.0.0.1的端口

  if ((time(NULL) - inst->lastquerytm) > 2) {
    /*printf("will query all udp port about topics:\n");
    for (int i = 0; i < info.count; i++)
    {
      printf("%s ", &info.topics[i][0]);
    }
    printf("\n");*/

    queryudpport(inst, &info);
    inst->lastquerytm = time(NULL);
  }
}
void golfdds_exit(GOLFDDS_NODE_HANDLE handle) {
  GOLFDDS_NODE_INSTANCE *inst = (GOLFDDS_NODE_INSTANCE *)handle;
  inst->willexit = 1;
}
void golfdds_spin(GOLFDDS_NODE_HANDLE handle) {
  GOLFDDS_NODE_INSTANCE *inst = (GOLFDDS_NODE_INSTANCE *)handle;
  // read from shm and then call callback
  while (!inst->willexit) {
    // printf("will work around subscriber to read!\n");
    // 遍历本节点的订阅者
    GOLFDDS_SUBSCRIBER_INSTANCE *tmpsub = inst->subscriber;
    while (tmpsub != NULL) {
      if (tmpsub->status == ATTACHED) { // 该订阅者已经处于ATTACHED状态，可以读取数据
        // printf("spin:find a attached topic:%s\n", tmpsub->topicname);
        //  read from shm
        uint8_t *p = (uint8_t *)SOLO_Read_SHM(tmpsub->msgqueue);
        if (p) {
          tmpsub->callback(p, tmpsub->msgsize);
          SOLO_Read_Over_SHM(tmpsub->msgqueue);
          // call callback
        } else {
          // printf("fail to read from queue of topic:%s\n", tmpsub->topicname);
        }
      } else {
        // printf("subscribe topic:%s,tobe detect!\n", tmpsub->topicname);
      }
      tmpsub = tmpsub->next;
    }
    // printf("will detect publisher!\n");
    detectpublisher(inst);
    usleep(2 * 1000);
  }
}
